package extract.document.general;

import com.fasterxml.jackson.databind.JsonNode;
import common.dao.couchdb.CouchDBDOCFactory;
import common.dao.elastic.ElasticSearchDao;
import common.helper.json.JsonToJavaObject;
import org.ektorp.CouchDbConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * Created by Common on 9/11/16.
 */
public abstract class BaseBatchMode {
    final Logger logger = LoggerFactory.getLogger(BaseBatchMode.class);

    public String indexName;
    public String typeName;

    private CouchDBDOCFactory couchdb;
    protected CouchDbConnector couch_con;

    private ElasticSearchDao esdao;

    public BaseBatchMode(CouchDBDOCFactory couchdb, ElasticSearchDao esdao){
        this.couchdb = couchdb;
        //this.couch_con = couchdb.;

        this.esdao = esdao;

        this.indexName = "media";
        this.typeName  = "video";

    }


    public void traversalCouchDB(){
        List<String> ids = this.couch_con.getAllDocIds();
        int index = 0;
        for(String couch_id : ids){
            // 1) get doc
            JsonNode doc = this.couch_con.get(JsonNode.class , couch_id.toString());
            logger.info("[traversalCouchDB] id = " + couch_id.toString() );
            logger.debug("[traversalCouchDB] doc = \n" + doc.toString());
            // 2) verify
            boolean isvalid = false;
            try{
                isvalid = verifyDoc(doc);
            }
            catch (Exception e){
                //e.printStackTrace();
                logger.error("[traversalCouchDB] verify failed! id = " + couch_id.toString() + ", doc = \n" + doc.toString());
                logger.error(e.toString());
                isvalid = false;
            }
            if(!isvalid){
                logger.error("[traversalCouchDB] invalid! ignore and continue!");
                continue;
            }
            // 3) transform
            JsonNode doc_index = transformToIndex(doc);
            saveToElastic(doc_index);

            // 4) add index
            index ++;
            logger.info("[traversalCouchDB] count = " + index);
        }
    }


    // will be override
    public boolean verifyDoc(JsonNode doc){

        return true;
    }


    // will be override
    public JsonNode transformToIndex(JsonNode input){
        //String input_json = JsonToJavaObject.jsonNodeToString(input);
        String input_json = input.toString();
        JsonNode output = JsonToJavaObject.loadJsonNode( input_json );
        return output;
    }

    // will be override, save_bulk to different type
    public boolean saveToElastic(JsonNode doc){
        this.esdao.save_bulk(this.indexName, this.typeName, doc);
        return true;
    }


}
